#include "ptwrap.hpp"
#include <deque>
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <assert.h>

struct Message 
{
	Message(int mtype) : msgtype_(mtype)
	{ }
	virtual ~Message() { } // allow for polymorphism

	enum {
		EXIT,
		CANCEL,
		NORMAL, // first normal type
		SOME_MSG_1
	};
	int msgtype_;
	// ... whatever else is needed by the application ...
};

// DEBUG {
class Queue
{
// copied from ex04cm
public:
	enum {
		DEFAULT_LIMIT = 1000
	};

	Queue(int limit = DEFAULT_LIMIT) :
		limit_(limit), 
		dqwr_(&deque1_), dqrd_(&deque2_), condrd_(condwr_),
		exitposted_(false), nwriters_(0)
	{ }
	virtual ~Queue()
	{ 
		dispose(deque1_);
		dispose(deque2_);
	}
	// ...
	virtual bool post(Message *m)
	{
		int res;
		{
			pw::lockmutex lm(mutexx_);
			res = post_l(m);
		}
		if (res == P_POSTIT)
			return push_back(m);
		else
			return (res != P_FAIL);
	}
	// ...
	Message *pop_front()
	{
		pw::lockmutex lm(mutexrd_);

		while(dqrd_->empty()) {
			{
				pw::lockmutex lmw(condwr_);

				if (!dqwr_->empty()) {
					swap(dqwr_, dqrd_);
					// if the write buffer was full, wake up the writers
					condwr_.broadcast();
				}
			}
			if(dqrd_->empty()) {
				pw::swapmutex sw(mutexrd_, condwr_);

				if (dqwr_->empty())
					condrd_.wait(); 
			}
		}

		Message *m = dqrd_->front();
		dqrd_->pop_front();
		return m;
	}
	// virtual in case if later a subclass wants to do something different;
	// returns true on success, or false if the handler has already been told
	// to exit
	virtual bool add_writer()
	{ 
		pw::lockmutex lm(mutexx_);
		if (exitposted_)
			return false;
		++nwriters_;
		return true;
	}
protected:
	enum post_l_result {
		P_FAIL,
		P_SUCCESS,
		P_POSTIT
	};

	virtual post_l_result post_l(Message *m)
	{
		bool pushlocked = false;
		if (exitposted_) {
			delete m;
			return P_FAIL;
		}
		switch (m->msgtype_) {
		case Message::CANCEL:
			clear(); // then the posting is done as usual...
			pushlocked = true;
			break;
		case Message::EXIT:
			if (--nwriters_ <= 0) {
				// last writer, post EXIT to the handler
				exitposted_ = true;
			} else {
				// message is processed successfully but it has
				// nothing to pass to the handler yet
				delete m;
				return P_SUCCESS;
			}
			pushlocked = true;
			break;
		// ...
		}
		if (pushlocked) {
			if (push_back(m)) // should never fail here, but just in case
				return P_SUCCESS;
			else
				return P_FAIL;
		} else {
			return P_POSTIT;
		}
	}

	// ...
	// Caller must have the mutex in condwr_ locked.
	bool push_back_l(Message *m)
	{
		if (exitposted_ && m->msgtype_ != Message::EXIT) {
			delete m;
			return false;
		}

		while (dqwr_->size() >= limit_)
			condwr_.wait(); // it releases and gets back the mutex

		dqwr_->push_back(m);
		condrd_.broadcast();
		return true;
	}
	// returns false if the exit was already posted, and this
	// is not the EXIT message itself (then the message also
	// gets discarded)
	bool push_back(Message *m)
	{
		pw::lockmutex lm(condwr_);
		return push_back_l(m);
	}
	void clear() 
	{
		pw::lockmutex lmr(mutexrd_);
		pw::lockmutex lmw(condwr_);

		dispose(*dqrd_);
		dispose(*dqwr_);

		// if the write buffer was full, wake up the writers
		condwr_.broadcast();
	}
	void dispose(std::deque<Message *> &deq)
	{ 
		Message *m;
		while (!deq.empty()) {
			m = deq.front();
			deq.pop_front();
			delete m;
		}
	}

	int limit_;
	std::deque<Message *> deque1_, deque2_;
	pw::pmutex mutexrd_;
	std::deque<Message *> *dqwr_, *dqrd_;
	pw::pmcond condwr_;
	pw::pchaincond condrd_;
	pw::pmutex mutexx_; // for the exit accounting
	bool exitposted_; // whether the EXIT message was posted to the queue
	int nwriters_; // number of writers that haven't already exited
};
// DEBUG }

// [[ex04db]]
// DEBUG descr Priority message processing in the context of the posting thread.
class SomeMessage : public Message
{
public:
	// DEBUG {
	SomeMessage(int mtype) : Message(mtype)
	{ }
	virtual int getData()
	{
		return -1;
	}
	// DEBUG }
	enum {
		// start after the base-class msg types
		URGENT_MSG_1 = NORMAL, 
		URGENT_MSG_2,
		SOME_NORMAL
	};
};

// DEBUG {
class Handler : public pw::pwthread
{
};

// DEBUG }
class SomeHandler : public Handler
{
	friend class SomeQueue;
public:
	// DEBUG {
	void setMyQueue(Queue *mine)
	{
		q_ = mine;
	}
	void setNextQueue(Queue *next)
	{
		nextq_ = next;
	}
	// DEBUG }
	// ...
	void *execute()
	{
		while(true) {
			SomeMessage *m = (SomeMessage *)q_->pop_front();
			{
				pw::lockmutex lm(mutex_);

				switch (m->msgtype_) {
				case Message::EXIT:
	// DEBUG {
					 fprintf(stderr, "Process exit\n");
	// DEBUG }
					nextq_->post(m);
					return 0;
				case Message::CANCEL:
					nextq_->post(m);
					break;
				case SomeMessage::SOME_NORMAL:
					// ... process it ...
	// DEBUG {
					 fprintf(stderr, "Start process normal %d\n", m->getData());
					sleep(1);
					 fprintf(stderr, "End process normal %d\n", m->getData());
	// DEBUG }
					delete m;
					break;
				}
			}
		}
	}

protected:
	Queue *q_;
	// input queue of the next handler in pipeline
	Queue *nextq_; 
	pw::pmutex mutex_;
};

class SomeQueue : public Queue
{
public:
	SomeQueue(SomeHandler *handler, 
			int limit = DEFAULT_LIMIT) :
		Queue(limit), handler_(handler)
	{ }

	virtual bool post(Message *m)
	{
		// note: no locking of mutexx_!
		switch (m->msgtype_) {
		case SomeMessage::URGENT_MSG_1:
			{
				pw::lockmutex lm(handler_->mutex_);
				// ... do the handling ...
	// DEBUG {
					 fprintf(stderr, "Process URGENT_1 %d\n", ((SomeMessage *)m)->getData());
	// DEBUG }
				delete m;
				return true;
			};
		case SomeMessage::URGENT_MSG_2:
			{
				pw::lockmutex lm(handler_->mutex_);
				bool res = true;
				// ... do the handling,may change res ...
	// DEBUG {
					 fprintf(stderr, "Process URGENT_2 %d\n", ((SomeMessage *)m)->getData());
	// DEBUG }
				delete m;
				return res;
			};
		default:
			return Queue::post(m);
		}
	}

protected:
	SomeHandler *handler_;
};
// END


// DEBUG {
struct data : public SomeMessage
{
	int v_;

	data(int v, int kind = SomeMessage::SOME_NORMAL) : SomeMessage(kind), v_(v)
	{ }
	~data()
	{
		fprintf(stderr, "deleted data %d\n", v_);
	}
	virtual int getData()
	{
		return v_;
	}
};

int main()
{
	SomeHandler h;
	SomeQueue q(&h);
	q.add_writer();

	Queue qnext;
	qnext.add_writer();

	h.setMyQueue(&q);
	h.setNextQueue(&qnext);

	h.start();
	sleep(1);

	q.post(new data(1));
	q.post(new data(2));
	q.post(new data(3, SomeMessage::URGENT_MSG_1));
	q.post(new data(99, Message::EXIT));
	h.join();

	return 0;
}
/* Sample output (timing of URGENT processing may vary):
Process URGENT_1 3
deleted data 3
Start process normal 1
End process normal 1
deleted data 1
Start process normal 2
End process normal 2
deleted data 2
Process exit
deleted data 99
*/
// DEBUG }
